package com.lmind.core.publish;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;

import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;

public class SocketPublisherRouterTestCase {
	
	private static ScheduledExecutorService scheduledExecutor;
	
	private static SimplePublisher<Serializable> publisher;
	
	private static SocketPublisherRouter<Serializable> server;
	
	private static SocketPublisherRouter<Serializable> client;
	
	@BeforeClass
	public static void beforeClass() throws Exception {
		
		scheduledExecutor = new ScheduledThreadPoolExecutor(5);
		
		publisher = new SimplePublisher<Serializable>();
		
		server = new SocketPublisherRouter<Serializable>();
		server.setPublisher(publisher);
		server.setScheduledExecutor(scheduledExecutor);
		server.bind(new NioSocketAcceptor(), new InetSocketAddress(21001));
		
		client = new SocketPublisherRouter<Serializable>();
		client.setScheduledExecutor(scheduledExecutor);
		client.connect(new InetSocketAddress("127.0.0.1", 21001));
		
		Thread.sleep(1000);
	}
	
	@AfterClass
	public static void afterClass() throws Exception {
		
		client.close();
		server.close();
		scheduledExecutor.shutdownNow();
	}
	

	@Test
	public void test() throws Exception {
		
		String news = "New York Times";
		final long loop = 1000;
		final Object lock = new Object();
		
		class MySubscriber implements Subscriber<Serializable> {
			private volatile long count = 0;
			@Override
			public void feed(String name, Serializable publication) {
				System.out.println(publication);
				count++;
				if (count == loop) {
					synchronized (lock) {
						lock.notify();
					}
				}
			}
		}
		
		MySubscriber sub = new MySubscriber();
		client.subscribe(sub, news);
		Thread.sleep(100);
		
		for (int i = 0; i < 1000; i++) {
			publisher.publish(news, "The World War! " + i);
		}
		
		synchronized (lock) {
			lock.wait(2000);
		}
		Assert.assertTrue(sub.count == loop);
	}
	
}
